package com.surfilter.massdata.spark.input.ipcheck;

import java.util.HashMap;
import java.util.Map;

import org.apache.commons.lang3.StringUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.DataFrameReader;

import com.act.sparkanalyz.input.ISparkInput;
import com.act.sparkanalyz.service.impl.SparkService.ServiceContext;
/** 
* @author  hutao
* @date 创建时间：2017年5月19日 下午5:27:17 
* @Description：备案来源库输入
* @version 1.0 
* 
**/
public class IpCheckBeianSourceInput implements ISparkInput {
	private static final long serialVersionUID = 1L;
	private static final Log log = LogFactory.getLog(IpCheckBeianSourceInput.class);
	private String driver;
	private String url;
	private String userName;
	private String password;
	private String beianSourceTableName;
	private String dfKey;
	private String beianSourceSql;
	@Override
	public Map<String, DataFrame> read(ServiceContext serviceContext, Map<String, String> commandMap) throws Exception {
		// TODO Auto-generated method stub
		log.info("==============IpCheckBeianSourceInput begin====================");
		Map<String, DataFrame> result = new HashMap<String, DataFrame>();
		try{
		DataFrame dataFrame = null;
		if (StringUtils.isNotBlank(beianSourceSql)) {
			Map<String, String> options = new HashMap<String, String>();
			options.put("url", url);
			options.put("driver", driver);
			options.put("dbtable", "(SELECT *  FROM ( " + beianSourceSql + " ) " + beianSourceTableName + ")");
			options.put("user", userName);
			options.put("password", password);
			DataFrameReader reader = serviceContext.getSqlCtx().read().format("jdbc").options(options);
			dataFrame = reader.load();
		}
//		dataFrame.registerTempTable(beianSourceTableName + "_register_temp");
//		dataFrame.persist(StorageLevel.MEMORY_AND_DISK_SER());
		result.put(dfKey, dataFrame);
		}catch(Exception e){
			if(log.isErrorEnabled()){
				log.error(e.getMessage(),e);
			}
		}
		log.info("==============IpCheckBeianSourceInput end====================");
		return result;
	}

}
